package com.wudl.flink.stream.tablesql

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table, TableEnvironment}
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors._

object TableKafka {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv = StreamTableEnvironment.create(env)
    tableEnv.connect(new Kafka().version("0.11")
      .topic("test")
      .property("zookeeper.connect", "10.204.125.141:2181")
      .property("bootstrap.servers", "10.204.125.141:9092")
    ).withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      ).createTemporaryTable("kafkatable")

//    kafkatableval kafkaTable: Table = tableEnv.from("kafkatable")
    val kafkaTable: Table = tableEnv.sqlQuery("select * from kafkatable where id = 'sensor_1' ")
    kafkaTable.toAppendStream[(String,Long,Double)].print()

    env.execute("kafkaTable-")


  }

}
